package com.hzh.SparkSQL

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo2DSLWC {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("sql")
      .getOrCreate()

    /**
     * 1、读取数据构建DataFrame，DF相当于一张表
     */
    val linesDF: DataFrame = spark
      .read
      .format("csv") //指定读取数据的文件格式
      .schema("line STRING") //指定字段名和字段类型
      .option("sep", "\t") //指定分隔符，csv格式默认逗号分隔
      .load("data/words.txt")

    /**
     * DSL:类sql api，介于代码和SQL之间的一种写法
     * 在写DSL之前需要导入sparkSql的函数和隐式转换
     */
    import org.apache.spark.sql.functions._
    import spark.implicits._


    //$"line" 获取列对象
    linesDF
      //将一行中的多个单词拆分成多行
      .select(explode(split($"line", ",")) as "word")
      //按照单词分组
      .groupBy($"word")
      //统计单词的数量
      .agg(count($"word") as "c")
      .write //保存数据
      .format("csv")
      .option("sep","\t")
      .mode(SaveMode.Overwrite)
      .save("data/wc2")


  }

}
